package channel

import (
	"gitee.com/lipore/go-nio/pkg/events"
	"gitee.com/lipore/go-nio/pkg/frame"
	"gitee.com/lipore/plume/logger"
	"sync"
	"time"
)

type (
	Pipeline struct {
		//handler list, the handler need to implement ConnectHandler, DisconnectHandler, InboundHandler, OutboundHandler,
		//ActiveHandler, InactiveHandler or ErrorHandler
		handlers []Handler

		// following pools are used to store the context object, to prevent excessive memory fragmentation when the context object allocated/released
		outboundPool   sync.Pool
		inboundPool    sync.Pool
		errorPool      sync.Pool
		connectPool    sync.Pool
		disconnectPool sync.Pool
		activePool     sync.Pool
		inactivePool   sync.Pool

		messageList chan inboundMessage // cache for message to be handled
		mu          sync.Mutex          //write lock for pCount
		pCount      int                 //the count of the message processor.
		codec       frame.CodeC         //frame coder that used by pipeline.
	}
	inboundMessage struct {
		c Channel
		f interface{}
	}
)

func NewPipeline(c frame.CodeC) *Pipeline {
	return &Pipeline{
		handlers: make([]Handler, 0),
		outboundPool: sync.Pool{
			New: func() interface{} {
				return &outboundContext{}
			},
		},
		inboundPool: sync.Pool{
			New: func() interface{} {
				return &inboundContext{}
			},
		},
		errorPool: sync.Pool{
			New: func() interface{} {
				return &errorContext{}
			},
		},
		connectPool: sync.Pool{
			New: func() interface{} {
				return &connectContext{}
			},
		},
		disconnectPool: sync.Pool{
			New: func() interface{} {
				return &disconnectContext{}
			},
		},

		messageList: make(chan inboundMessage, 50),
		codec:       c,
	}
}

func (p *Pipeline) Size() int {
	return len(p.handlers)
}

func (p *Pipeline) AddFirst(handlers ...Handler) {
	p.handlers = append(handlers, p.handlers...)
}

func (p *Pipeline) AddLast(handlers ...Handler) {
	p.handlers = append(p.handlers, handlers...)
}

func (p *Pipeline) AddHandlers(position int, handlers ...Handler) {
	if position == 0 {
		p.AddFirst(handlers...)
	} else if position >= p.Size() {
		p.AddLast(handlers...)
	} else {
		h := append(p.handlers[:position], handlers...)
		p.handlers = append(h, p.handlers[position:]...)
	}
}

func (p *Pipeline) FireChannelInbound(channel Channel) {
	buffer := channel.InboundBuffer()
	for {
		if f, has, err := p.codec.Decode(buffer); has {
			p.enqueueMessage(channel, f)
		} else if err != nil {
			logger.Warnf("decode failed: %v", err)
			_ = channel.Close()
			return
		} else {
			break
		}
	}
}

func (p *Pipeline) FireChannelEvent(channel Channel, event *events.EventWithCode) {
	p.enqueueMessage(channel, event)
}

func (p *Pipeline) enqueueMessage(c Channel, m interface{}) {
	if len(p.messageList) >= 25 || p.pCount == 0 {
		go channelInBound(p)
		p.mu.Lock()
		p.pCount++
		p.mu.Unlock()
	}
	p.messageList <- inboundMessage{c: c, f: m}
}

func (p *Pipeline) FireChannelOutbound(channel Channel, message frame.Frame) {
	ctx := p.outboundPool.Get().(OutboundContext)
	ctx.reset(channel, p.handlers)
	ctx.bindRaw(message)
	err := ctx.Next()
	if err != nil {
		_ = channel.Close()
		return
	}
	if f := ctx.frame(); f != nil {
		if packet, ok := p.codec.Encode(f); ok {
			_, err = ctx.transportWrite(packet)
			if err != nil {
				logger.Debugf("%v", err)
				return
			}
		}
	}
	p.outboundPool.Put(ctx)
}

func channelInBound(p *Pipeline) {
	timer := time.NewTimer(1 * time.Minute)
	for {
		select {
		case <-timer.C:
			p.mu.Lock()
			p.pCount--
			p.mu.Unlock()
			return
		case m := <-p.messageList:
			switch t := m.f.(type) {
			case frame.Frame:
				handleInbound(p, m.c, t)
			case *events.EventWithError:
				handleError(p, m.c, t)
			case *events.EventWithCode:
				switch t {
				case events.ConnectEvent:
					handleConnect(p, m.c)
				case events.DisconnectEvent:
					handleDisconnect(p, m.c)
				case events.ActiveEvent:
					handleActive(p, m.c)
				case events.InactiveEvent:
					handleInactive(p, m.c)
				}
			}
			timer.Reset(time.Minute)
		}
	}
}

func handleInbound(p *Pipeline, c Channel, f frame.Frame) {
	ctx := p.inboundPool.Get().(InboundContext)
	ctx.reset(c, p.handlers)
	ctx.bindFrame(f)
	err := ctx.Next()
	if err != nil {
		ctx.Disconnect()
	}
	p.inboundPool.Put(ctx)
}
func handleError(p *Pipeline, c Channel, err error) {
	ctx := p.errorPool.Get().(ErrorContext)
	ctx.reset(c, p.handlers)
	ctx.bindErr(err)
	e := ctx.Next()
	if e != nil {
		ctx.Disconnect()
	}
	p.errorPool.Put(ctx)
}
func handleConnect(p *Pipeline, c Channel) {
	ctx := p.connectPool.Get().(ConnectContext)
	ctx.reset(c, p.handlers)
	err := ctx.Next()
	if err != nil {
		ctx.Disconnect()
	}
	p.connectPool.Put(ctx)
}
func handleDisconnect(p *Pipeline, c Channel) {
	ctx := p.disconnectPool.Get().(DisconnectContext)
	ctx.reset(c, p.handlers)
	err := ctx.Next()
	if err != nil {
		ctx.Disconnect()
	}
	p.disconnectPool.Put(ctx)
}

func handleActive(p *Pipeline, c Channel) {
	ctx := p.activePool.Get().(ActiveContext)
	ctx.reset(c, p.handlers)
	err := ctx.Next()
	if err != nil {
		ctx.Disconnect()
	}
	p.activePool.Put(ctx)
}

func handleInactive(p *Pipeline, c Channel) {
	ctx := p.inactivePool.Get().(InactiveContext)
	ctx.reset(c, p.handlers)
	err := ctx.Next()
	if err != nil {
		ctx.Disconnect()
	}
	p.inactivePool.Put(ctx)
}
